package com.spring.demo.webflux.demo.controller;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 、SSE 长轮询实现 Server Sent Event (服务器推送事件)  单向消息传递
 *
 * @author wangzhilong
 * @date 2021/3/2
 */
@RestController
public class SSEController {

    private static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();

    private Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 3.1版本 sse 改进了返回值：不需要 data: 数据  /n/n 格式
     * 3.1版本 sse 是同一个线程处理推送
     * @return
     */
    @RequestMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Object push() {
        log.info("老铁，你来啦，thread:{}", Thread.currentThread().getName());
        final LocalDateTime now = LocalDateTime.now();
        return now.toString() ;
    }

    /**
     * 该方法验证失败,web端没有吼住请求，请求是一直在发（问题产生的原因是当前使用的是reactive 模块），spring mv 则不存在该问题
     * @param id
     * @return
     */
    @GetMapping("/sse/subscribe")
    public SseEmitter subscribe(String id) {
        //超时时间设置为1小时
        SseEmitter sseEmitter = new SseEmitter(3600000L);
        sseCache.put(id, sseEmitter);
        //超时回调触发
        sseEmitter.onTimeout(() -> {
            System.out.println("超时回调触发，清除ID");
            sseCache.remove(id);
        });
        //结束之后的回调触发
        sseEmitter.onCompletion(() -> System.out.println("完成！"));
        return sseEmitter;
    }

    @GetMapping("sse/push")
    public String push(String id,String context) throws IOException {
        final SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            sseEmitter.send(context);
        }
        return "over";
    }

    @GetMapping("sse/pushAll")
    public String push(String context) {
        sseCache.values().stream().forEach(e -> {
            try {
                e.send(context);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        });
        return "over";
    }



}
